package com.gugee.ins.data.collector.blogger.listener;

import com.gugee.ins.data.collector.blogger.config.CollectorBloggerConfig;
import com.gugee.ins.data.collector.blogger.service.InsBloggerService;
import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 博主收入监听器
 * @author Huangxin
 */
@Component
@Slf4j
public class InsBloggerUpdateListener {
    /**
     * 配置类
     */
    @Autowired
    CollectorBloggerConfig collectorBloggerConfig;

    /**
     * insBlogger服务
     */
    @Autowired
    InsBloggerService insBloggerService;

    private InsBloggerUpdateHandler collectHandler = new InsBloggerUpdateHandler();

    /**
     * 监听任务
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insBloggerUpdate.id}",topics = {"${kafka.topics.insBloggerUpdate.online}"},containerFactory = "factoryInsBloggerUpdateAck")
    public void bloggerMainCollect(List<InsBloggerMessage> messages, Acknowledgment ack) {
        collectHandler.handle("InsBloggerCollectUpdate",messages, collectorBloggerConfig.getAcceptInsBloggerUpdateMsgTimeMinRange(),ack);
    }
    @KafkaListener(id="${kafka.consumer.group.insBloggerUpdatePriority.id}",topics = {"${kafka.topics.insBloggerUpdate.priority}"},containerFactory = "factoryInsBloggerUpdatePriorityAck")
    public void bloggerMainCollectPriority(List<InsBloggerMessage> messages, Acknowledgment ack) {
        collectHandler.handle("InsBloggerCollectUpdatePriority",messages, collectorBloggerConfig.getAcceptInsBloggerUpdateMsgTimeMinRange(),ack);
    }

    class InsBloggerUpdateHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage> {
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insBloggerService.handleInsBloggerUpdate(messages);
        }
    }
}
